package com.shujia.flink.source

import org.apache.flink.streaming.api.scala._

object Demo1Source {
  def main(args: Array[String]): Unit = {

    //创建环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /**
      * 基于本地集合的source --- 有界流
      * 数据处理完任务结束
      *
      */

    val listDS: DataStream[Int] = env.fromCollection(List(1, 2, 3, 4, 5, 6, 7, 8, 9))

    listDS.print()


    /**
      * 基于文件的source  --- 有界流
      *
      */

    val studentsDS: DataStream[String] = env.readTextFile("flink/data/students.txt")

    //统计班级人数
    //每一条数据都会计算出结果
    studentsDS
      .map(line => line.split(",")(4))
      .map((_, 1))
      .keyBy(_._1)
      .reduce((x, y) => (x._1, x._2 + y._2))
      .print()


    /**
      * 基于socket source  -- 无界流
      *
      * 只要代码中有一个无界流代码就不会结束
      *
      */


    val socketDS: DataStream[String] = env.socketTextStream("master", 8888)

    socketDS.print()

    env.execute()

  }

}
